import redis
import pickle
import configparser
import logging
import pandas as pd
import datetime
from sqlalchemy import create_engine, DateTime, String
import pymysql
import time
import numpy as np
from send_email import sendMessage
import traceback

pymysql.install_as_MySQLdb()


log_format = "%(asctime)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s"
date_format = "%Y-%m-%d %H:%M:%S"  # 精确到秒
logging.basicConfig(level=logging.DEBUG, format=log_format, datefmt=date_format)

# 初始化配置解析器
config = configparser.ConfigParser()

# 读取配置文件
import os
current_dir = os.path.dirname(os.path.abspath(__file__))
config.read(current_dir+'/config.ini')


# 获取Redis的配置信息
redis_host = config.get('Redis', 'host')
redis_port = config.getint('Redis', 'port')
redis_db = config.getint('Redis', 'db')
redis_password = config.get('Redis', 'password')
r = redis.Redis(host=redis_host, port=redis_port, db=redis_db, password=redis_password)

mysql_port = config.getint('mysql', 'port')
mysql_host = config.get('mysql', 'host')
mysql_db = config.get('mysql', 'db')
import urllib.parse
mysql_password = urllib.parse.quote(config.get('mysql', 'password'))
mysql_user = config.get('mysql', 'user')
db_url = f'mysql://{mysql_user}:{mysql_password}@{mysql_host}:{mysql_port}/{mysql_db}'

engine = create_engine(db_url,pool_size=20,max_overflow=20,pool_recycle=60)


def get_real_stock_to_db():
    if r.get("is_trade_time") ==b"NO":
        return
    
    values = []
    for key in r.scan_iter("real_stock_info:*"):
        values.append(pickle.loads(r.get(key)))
    df = pd.concat(values)


    df["change"] = round((df["price"]-df["last_close"])*100/df["last_close"],2)
    df["ask_money"] = df["ask1"] * df["ask_vol1"] * 100 / 10000
    df["sell_money"] = df["bid1"] * df["bid_vol1"] * 100 / 10000
    df = df[["code","price","vol","cur_vol","s_vol","b_vol","amount","change","open","last_close","high","low","ask_money","sell_money"]]


    df['timestamp']  = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    df['avr_change'] = round((df['amount']/100/df["vol"]-df['last_close'])*100/df['last_close'],2)
    df['amount']  = round(df['amount']/10000,2)
    df['high'] = round((df['high']-df['price'])*100/df['price'],2)
    df['low'] = round((df['price']-df['low'])*100/df['price'],2)

    df = df.drop_duplicates()
    df.replace([np.inf, -np.inf], 0, inplace=True)
    df["ask_money"] = df["ask_money"].replace([np.inf, -np.inf], 0)
    df["sell_money"] = df["sell_money"].replace([np.inf, -np.inf], 0)

    logging.info(f"保存数据{len(df)}条")
    r.set("real_stock_info_tdx",pickle.dumps(df))
    # r.expire('real_stock_info_tdx', 3600*18)
    
    df.to_sql("real_market_info_tdx", engine, if_exists='append', index=False, dtype={'timestamp': DateTime(),'code': String(length=8)})

while True:
    try:
        start_time = time.time()
        get_real_stock_to_db()
        end_time = time.time()
        elapsed_time = end_time - start_time
        sleep_time = max(1 - elapsed_time, 0)
        time.sleep(sleep_time)
    except Exception as e:
        tb_info = traceback.format_exc()
        logging.info(f"An error occurred: {e}\nTraceback info:\n{tb_info}")
        logging.error(e)
        time.sleep(1)
        # sendMessage("整理通达信实时数据错误，请检查")